package rabbitmq.rpc;

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class Client {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("110.41.84.205");
        factory.setPort(5672); // 端口号
        factory.setUsername("study"); // 账号
        factory.setPassword("study"); // 密码
        factory.setVirtualHost("Demo"); // 虚拟主机

        Connection connection = factory.newConnection();// 建立连接
        // 建立通道
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(Constants.RPC_QUEUE_RPC,true, false, false,  null);
        channel.queueDeclare(Constants.RPC_QUEUE_REPLY,true, false, false,  null);
        String correlationId = UUID.randomUUID().toString();

        // 设置配置
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                .correlationId(correlationId)
                .replyTo(Constants.RPC_QUEUE_REPLY)
                .build();
        // 发送消息
        String msg = "hello rpc通信";
        channel.basicPublish("",Constants.RPC_QUEUE_RPC, true, false,  properties, msg.getBytes());
        System.out.println("发送成功");



        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        // 接收请求
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envellope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                if (properties.getCorrelationId().equals(correlationId)) {
                    response.offer(new String(body, "utf-8"));
                    // 确认应答
                    channel.basicAck(envellope.getDeliveryTag(), false);
                }
            }

        };
        channel.basicConsume(Constants.RPC_QUEUE_REPLY, consumer);
        String result = response.take();
        System.out.println("[响应] result： " + result);
        channel.close();
        connection.close();
    }
}
